use crate::contract::coolink_sdk::{SDKApiResult, SGWebSocketCallBack, SGWebSocketContract};

use crate::imp::ws::websocket_imp::{
    MessageData, MessageToReceived, SendData, WebSocketImp, WebSocketInnerTrait,
    WebSoketEventCallback, WsSendMessage,
};
use crate::model::sdk_api_result::{
    SDKResultError, DEFAULT_CONNECTION_TIMEOUT_SECS, DEFAULT_REQUEST_TIMEOUT_SECS,
};
use crate::model::sdk_req_result::ReqResult;
use crate::utils::ids;
use crate::utils::logutil;
use crate::utils::mpsc_utils;
use crate::utils::query;
use anyhow::anyhow;
use futures_channel::mpsc::UnboundedSender;

use std::time::Duration;
// use rand::random;
// use crate::imp::rtm_sdk_imp::{call_attach_fn, call_detach_fn};
use serde::{Deserialize, Serialize};
use state::Storage;
use std::sync::RwLock;
use tokio::runtime::Builder;
use url::Url;

lazy_static! {
    static ref SG_WS_SDK_INSTANCE: Storage<RwLock<SGWebSocketContractImp>> = state::Storage::new();
    static ref IS_CONNECT_WS_RUN: Storage<RwLock<bool>> = state::Storage::new();
    static ref MY_GLOBAL_SOIRCE_ID: Storage<RwLock<String>> = state::Storage::new();
}

struct CustomWsCallback {
    callback: Box<dyn SGWebSocketCallBack>,
}

impl CustomWsCallback {
    pub fn new(callback: Box<dyn SGWebSocketCallBack>) -> Self {
        CustomWsCallback { callback }
    }
}

impl WebSoketEventCallback for CustomWsCallback {
    fn on_ws_event(
        &self,
        from_target_id: String,
        event: String,
        data: String,
        msg_id: String,
    ) -> Result<(), anyhow::Error> {
        let data = MessageToReceived {
            target_id: from_target_id,
            event,
            data,
            id: msg_id,
        };

        // on_sg_message received data
        self.callback
            .on_sg_message(serde_json::to_string(&data).unwrap_or_default());
        Ok(())
    }

    fn on_connect_event(
        &self,
        source_id: String,
        turn_addr: String,
        turn_usr_name: String,
        turn_usr_pwd: String,
        sender: Option<UnboundedSender<WsSendMessage>>,
    ) -> Result<(), anyhow::Error> {
        set_my_source_id(source_id.to_owned());
        let online = sender.is_some();
        // put the new sender into mpsc_utils.SENDER_WATCH.0
        if let Err(e) = mpsc_utils::set_sender_watch(sender) {
            log::warn!("on_connect_event set_sender_watch: fail {:?}", e);
        }
        // if option sender is ok,return online
        if online {
            self.callback
                .on_sg_online(source_id, turn_addr, turn_usr_name, turn_usr_pwd);
        }
        Ok(())
    }

    fn on_error_event(&self, msg: String) -> Result<(), anyhow::Error> {
        self.callback.on_sg_error(msg);
        Ok(())
    }

    fn on_ping_event(&self) -> Result<(), anyhow::Error> {
        self.callback.on_sg_ping();
        Ok(())
    }

    fn on_need_reconnect(&self, msg: String) -> Result<(), anyhow::Error> {
        self.callback.on_sg_need_reconnect(msg);
        Ok(())
    }

    fn on_send_result(
        &self,
        event_id: String,
        is_ok: bool,
        reason: String,
    ) -> Result<(), anyhow::Error> {
        self.callback.on_sg_send_reply(event_id, is_ok, reason);
        Ok(())
    }
}

pub struct InnerSGWebSocketContractImp;

impl InnerSGWebSocketContractImp {
    pub fn new() -> Self {
        InnerSGWebSocketContractImp {}
    }

    async fn async_connect(
        &self,
        server_host: String,
        subgroup_path: String,
        server_ak: String,
        server_secret: String,
        device_package: String,
        device_id: String,
        user_id: String,
        callback: Box<dyn SGWebSocketCallBack>,
        config_file_saved_path: String,
        is_server: bool,
    ) -> Result<(), anyhow::Error> {
        let mut websocket = WebSocketImp::new(
            server_host,
            subgroup_path,
            server_ak,
            server_secret,
            device_package,
            device_id,
            user_id,
            config_file_saved_path,
            is_server,
        )?;
        log::info!("InnerSGWebSocketContractImp =: {}", websocket);

        let my_callback = CustomWsCallback::new(callback);
        let my_callback_box = Box::new(my_callback);
        let id = websocket.add_callback(vec![], my_callback_box);
        websocket.connect().await?; // block here until websocket connect return
        websocket.remove_callback(&id);
        Ok(())
    }

    fn sync_connect(
        &self,
        server_host: String,
        subgroup_path: String,
        server_ak: String,
        server_secret: String,
        device_package: String,
        device_id: String,
        user_id: String, //use for account changed
        callback: Box<dyn SGWebSocketCallBack>,
        is_server: bool,
        config_file_saved_path: String,
    ) -> SDKApiResult {
        if let Ok(rt) = Builder::new_multi_thread()
            .on_thread_start(|| {
                // call_attach_fn();
            })
            .on_thread_stop(|| {
                // call_detach_fn();
            })
            .enable_all()
            .build()
        {
            // block on tokio async loop. sync_connect must be called in a thread!
            let res = rt.block_on(self.async_connect(
                server_host,
                subgroup_path,
                server_ak,
                server_secret,
                device_package,
                device_id,
                user_id,
                callback,
                config_file_saved_path,
                is_server,
            ));
            log::info!("websocekt connect end res: {:?}", &res);
            if !set_connect_state(false) {
                log::error!("sync_connect exit blockon,but set_connect_state return false and is_connect_state_true = {:?}!",is_connect_state_true());
            }
            // quit websocket connect beacuse close or some err,
            // so ,our can retry to
            if let Err(e) = res {
                // -1 mean library error!;
                log::info!("websocekt connect end error msg: {:?}", e);
                SDKApiResult::error_string(e.to_string())
            } else {
                SDKApiResult::ok()
            }
        } else {
            if !set_connect_state(false) {
                log::error!("sync_connect exit runtime error set_connect_state return false and is_connect_state_true = {:?}!",is_connect_state_true());
            }
            SDKApiResult::from(anyhow!(SDKResultError::ErrorRuntime))
        }
    }

    fn send_target_with_unique_id(
        &self,
        target_unique_id: String,
        event_name: String,
        data: String,
        qos: i32, //default = 0
                  // sendcb: Box<dyn SendResultCallback>,
    ) -> SDKApiResult {
        if MY_GLOBAL_SOIRCE_ID.try_get().is_none() {
            let fail = SDKApiResult::from(anyhow!(SDKResultError::ErrorNotClientSourceIDFound));
            // let c_id = sendcb.on_callback_id();
            // sendcb.on_sg_send_reply(c_id.to_string(), fail.code, fail.msg);
            return fail;
        }
        // qos support 0 (default) or 1
        // 0 means not delay,send immediately then target may be offline or not ack fai!
        let mut _send_qos = if qos != 0 || qos != 1 { 0 } else { qos };

        let event_id = ids::uuid().unwrap_or(ids::id_generate_xbit(6));
        log::info!(
            "send_target_with_unique_id target_unique_id:{}, event_id: {} , event_name: {}, data: {}",
            target_unique_id,
            event_id,
            event_name,
            data
        );
        // let t_out = if timeout <= 0 || timeout > 10 {
        //     10
        // } else {
        //     timeout
        // };

        // let unique_source_id = MY_GLOBAL_SOIRCE_ID.get().read().unwrap();
        // oneshot_utils::add_sender_map(event_id.to_owned(), sendcb);
        if let Err(e) = mpsc_utils::send(WsSendMessage::MessageToSend(SendData {
            target_id: target_unique_id,
            msg: MessageData {
                id: event_id.to_owned(),
                event: event_name,
                data,
            },
        })) {
            log::error!("send_data error: {:?}", e);
            let fail = SDKApiResult::from(anyhow!(SDKResultError::ErrorSendChannelBroken));
            // oneshot_utils::remove_oneshot_map(event_id.to_owned(), fail.code, fail.msg);
            return fail;
        }
        SDKApiResult::msg(event_id.to_owned())
    }

    fn disconnect(&self) -> SDKApiResult {
        // if had mpsc_utils::set_sender_watch,
        // and use the send channel to send Disconnect cmd
        // otherwise just return ok.
        if let Err(e) = mpsc_utils::send(WsSendMessage::Disconnect) {
            log::error!("disconnect send error: {:?}", e);
            return e.into();
        }
        SDKApiResult::ok()
    }
}

pub struct SGWebSocketContractImp {
    pub ws_instance: InnerSGWebSocketContractImp,
}

// set sdk instance only once
fn set_sdk_instance(sg_websocket_imp: SGWebSocketContractImp) -> bool {
    // if try get success from state::Storage
    let ok = if let Some(_) = SG_WS_SDK_INSTANCE.try_get() {
        // if sdk instance call many times in many thread and this api will return false.
        log::warn!("signal gateway sdk instance already inited!");
        false
    } else {
        // set new connect_state
        SG_WS_SDK_INSTANCE.set(RwLock::new(sg_websocket_imp));
        true
    };
    ok
}

// set connect state multi thread
pub fn set_connect_state(connect_state: bool) -> bool {
    let ok = if let Some(ws_state) = IS_CONNECT_WS_RUN.try_get() {
        if let Ok(mut ws_state) = ws_state.try_write() {
            // connect state can write
            if *ws_state != connect_state {
                *ws_state = connect_state;
                true
            } else {
                false
            }
        } else {
            false
        }
    } else {
        // set new connect_state
        IS_CONNECT_WS_RUN.set(RwLock::new(connect_state));
        true
    };
    return ok;
}

pub fn is_connect_state_true() -> bool {
    let is_connect = if let Some(is_connect) = IS_CONNECT_WS_RUN.try_get() {
        if let Ok(is_connect) = is_connect.try_read() {
            *is_connect
        } else {
            false
        }
    } else {
        false
    };
    log::info!("is_ws_has_connect {}", is_connect);
    is_connect
}

fn set_my_source_id(id: String) -> bool {
    if let Some(source_id) = MY_GLOBAL_SOIRCE_ID.try_get() {
        if let Ok(mut source) = source_id.try_write() {
            *source = id;
            return true;
        }
    } else {
        MY_GLOBAL_SOIRCE_ID.set(RwLock::new(id));
        return true;
    }
    return false;
}

impl SGWebSocketContractImp {
    pub fn new() -> Self {
        SGWebSocketContractImp {
            ws_instance: InnerSGWebSocketContractImp::new(),
        }
    }
}

#[derive(Serialize, Deserialize)]
struct QueryTargetIdData {
    // client_id
    #[serde(rename = "client_id")]
    client_id: String,
}

#[derive(Serialize, Deserialize)]
struct QueryTargetOnlineData {
    // is_online
    #[serde(rename = "online_status")]
    online_status: i8,
}

impl SGWebSocketContractImp {
    fn query_is_online(
        server_host: String,
        subgroup_path: String,
        server_ak: String,
        server_secret: String,
        client_id: String,
    ) -> Result<QueryTargetOnlineData, anyhow::Error> {
        if client_id == "" || server_host == "" {
            let msg = "sync_query_is_online client_id or host is empty!";
            log::info!("{}", msg);
            anyhow::bail!(msg);
        }

        if server_ak == "" || server_secret == "" {
            let msg = "sync_query_is_online app_key or secret empty!";
            log::info!("{}", msg);
            anyhow::bail!(msg);
        }

        log::info!(
            "sync_query_is_online subgrouphost: {} ,client_id:{}",
            server_host,
            client_id
        );

        let mut query: Vec<(String, String)> = vec![("client_id".to_owned(), client_id.to_owned())];
        let query_str =
            query::assemble_query_config((server_ak.as_str(), server_secret.as_str()), &mut query)?;

        let has_prefix =
            if server_host.starts_with("http://") || server_host.starts_with("https://") {
                true
            } else {
                false
            };
        let query_url = if has_prefix {
            format!("{}{}?{}", server_host, subgroup_path, query_str)
        } else {
            format!("https://{}{}?{}", server_host, subgroup_path, query_str)
        };

        log::info!("sync_query_is_online query_url: {}", query_url);
        // if url is not a illegal url ,then return.
        Url::parse(&query_url)?;

        let client = reqwest::blocking::Client::builder()
            .connect_timeout(Duration::from_secs(DEFAULT_CONNECTION_TIMEOUT_SECS))
            .gzip(true)
            // .trust_dns(true)
            .build()?;

        let body = client
            .get(query_url)
            .timeout(std::time::Duration::from_secs(DEFAULT_REQUEST_TIMEOUT_SECS))
            .send()?
            .text()?;
        let ReqResult { code, data, msg } = serde_json::from_str(&body)?;
        if code == 0 {
            match data {
                Some(query_data) => {
                    let query_data_res = serde_json::from_value(query_data)?;
                    Ok(query_data_res)
                }
                _ => anyhow::bail!("sync_query_is_online fail,code code = {}", code),
            }
        } else {
            if msg.is_some() {
                anyhow::bail!(
                    "sync_query_is_online fail,msg.is_some() code = {},msg = {}",
                    code,
                    msg.unwrap()
                );
            } else {
                anyhow::bail!("sync_query_is_online fail,msg.is_none() code = {}", code);
            }
        }
    }
    fn query_target_unqiue_id(
        server_host: String,
        subgroup_path: String,
        server_ak: String,
        server_secret: String,
        t_device_package: String,
        t_device_id: String,
        t_user_id: String,
    ) -> Result<QueryTargetIdData, anyhow::Error> {
        if t_device_id == "" || server_host == "" || t_device_package == "" {
            let msg = "sync_query_target_unqiue_id device_id or host or device_package is empty!";
            log::info!("{}", msg);
            anyhow::bail!(msg);
        }
        if server_ak == "" || server_secret == "" {
            let msg = "sync_query_target_unqiue_id app_key or secret empty!";
            log::info!("{}", msg);
            anyhow::bail!(msg);
        }
        log::info!(
            "sync_query_target_unqiue_id subgrouphost: {} ,device_id:{},device_pkg:{}",
            server_host,
            t_device_id,
            t_device_package
        );
        let mut query: Vec<(String, String)> = vec![
            ("device_id".to_owned(), t_device_id.to_owned()),
            ("device_pkg".to_owned(), t_device_package.to_owned()),
            ("uid".to_owned(), t_user_id.to_owned()),
        ];
        let query_str =
            query::assemble_query_config((server_ak.as_str(), server_secret.as_str()), &mut query)?;

        let has_prefix =
            if server_host.starts_with("http://") || server_host.starts_with("https://") {
                true
            } else {
                false
            };
        let query_url = if has_prefix {
            format!("{}{}?{}", server_host, subgroup_path, query_str)
        } else {
            format!("https://{}{}?{}", server_host, subgroup_path, query_str)
        };

        log::info!("sync_query_target_unqiue_id query_url: {}", query_url);
        // if url is not a illegal url ,then return.
        Url::parse(&query_url)?;

        let client = reqwest::blocking::Client::builder()
            .connect_timeout(Duration::from_secs(DEFAULT_CONNECTION_TIMEOUT_SECS))
            .gzip(true)
            // .trust_dns(true)
            .build()?;

        let body = client
            .get(query_url)
            .timeout(std::time::Duration::from_secs(DEFAULT_REQUEST_TIMEOUT_SECS))
            .send()?
            .text()?;

        let ReqResult { code, data, msg } = serde_json::from_str(&body)?;
        if code == 0 {
            match data {
                Some(query_data) => {
                    let query_data_res = serde_json::from_value(query_data)?;
                    Ok(query_data_res)
                }
                _ => anyhow::bail!("sync_query_target_unqiue_id fail,code code = {}", code),
            }
        } else {
            if msg.is_some() {
                anyhow::bail!(
                    "sync_query_target_unqiue_id fail,msg.is_some() code = {},msg = {}",
                    code,
                    msg.unwrap()
                );
            } else {
                anyhow::bail!(
                    "sync_query_target_unqiue_id fail,msg.is_none() code = {}",
                    code
                );
            }
        }
    }
}

impl SGWebSocketContract for SGWebSocketContractImp {
    // "OFF", "ERROR", "WARN", "INFO", "DEBUG", "TRACE"
    fn init_sg_sdk(tag: String, log_level: String) -> SDKApiResult {
        if set_sdk_instance(SGWebSocketContractImp::new()) {
            logutil::global_log_init(tag.as_str(), log_level.as_str());
        }
        SDKApiResult::ok()
    }

    /// sync connect with websocket
    /// application layer must be spwan a thread
    /// if connect success will return unique device_id.
    fn loop_block_on_connect_sg(
        sg_host: String,                           // gateway url host address
        sg_subgroup_path: String,                  // gateway url subgroup path
        sg_access_key: String,                     // gateway registed http request auth access key
        sg_access_secret: String, // gateway registed http request auth access secret,corresponding for access key
        local_app_package: String, // gateway registed app package name
        local_device_id: String,  // regist to gateway local unique device id
        local_user_id: String,    // regist to gateway local unique user id
        sg_callback: Box<dyn SGWebSocketCallBack>, // function call back
        config_file_saved_path: String, //config file saved path.
    ) -> SDKApiResult {
        // safe set state in multithread.
        if !set_connect_state(true) {
            log::error!(
                "sync_connec set_connect_state in multithread,one only need call once,
            please handle the 40002 error and ignore this error result!"
            );
            return SDKApiResult::from(anyhow!(SDKResultError::ErrorAlreadyConnected));
        }
        match SG_WS_SDK_INSTANCE.try_get() {
            Some(instance) => {
                return instance.read().unwrap().ws_instance.sync_connect(
                    sg_host,
                    sg_subgroup_path,
                    sg_access_key,
                    sg_access_secret,
                    local_app_package,
                    local_device_id,
                    local_user_id,
                    sg_callback,
                    false,
                    config_file_saved_path,
                );
            }
            None => {
                if !set_connect_state(false) {
                    log::error!("sync_connect exit SG_WS_SDK_INSTANCE.get().read() error set_connect_state return false and is_connect_state_true = {:?}!",is_connect_state_true());
                }
                return SDKApiResult::from(anyhow!(SDKResultError::ErrorLinkSdkNotInited));
            }
        };
    }
    /// send message to server_unique_id exits in server within same ak,
    /// if A run sync_ws_connection and return unique_id,
    /// then B can call this api with a:unique_id
    /// the result return some about offline.
    fn send_message_sg_relay_to(
        unique_client_id: String, // target unique_client_id
        event_name: String,       // message event name
        event_data: String,       // message event data
        qos: i32,                 // message qos level,default = 0
    ) -> SDKApiResult {
        match SG_WS_SDK_INSTANCE.try_get() {
            Some(instance) => {
                return instance
                    .read()
                    .unwrap()
                    .ws_instance
                    .send_target_with_unique_id(unique_client_id, event_name, event_data, qos);
            }
            None => {
                // fail
                let result = SDKApiResult::from(anyhow!(SDKResultError::ErrorLinkSdkNotInited));
                return result;
            }
        };
    }

    // fn gen_send_cb_id() -> i64 {
    //     let r_i64 = random::<i64>();
    //     r_i64
    // }

    fn block_query_sg_client_id(
        sg_host: String,          // gateway query clientid url host address
        sg_subgroup_path: String, //  gateway query clientid url subgroup path
        sg_access_key: String,    // gateway registed http request auth access key
        sg_access_secret: String, // gateway registed http request auth access secret,corresponding for access key
        app_package: String,      // gateway registed app package name
        device_id: String,        // regist to gateway local unique device id
        user_id: String,          // regist to gateway local unique user id
    ) -> SDKApiResult {
        match SGWebSocketContractImp::query_target_unqiue_id(
            sg_host,
            sg_subgroup_path,
            sg_access_key,
            sg_access_secret,
            app_package,
            device_id,
            user_id,
        ) {
            Ok(item) => SDKApiResult::msg(item.client_id),
            Err(e) => e.into(),
        }
    }

    fn block_query_sg_is_online(
        sg_host: String,          // gateway query client is online url host address
        sg_subgroup_path: String, //  gateway query client is online url subgroup path
        sg_access_key: String,    // gateway registed http request auth access key
        sg_access_secret: String, // gateway registed http request auth access secret,corresponding for access key
        sg_client_id: String,     // regist to gateway unique client id
    ) -> SDKApiResult {
        match SGWebSocketContractImp::query_is_online(
            sg_host,
            sg_subgroup_path,
            sg_access_key,
            sg_access_secret,
            sg_client_id,
        ) {
            Ok(item) => SDKApiResult::msg(if item.online_status == 1 {
                "1".to_string()
            } else {
                "0".to_string()
            }),
            Err(e) => e.into(),
        }
    }
    /// check if websocket has connect
    fn had_connected_sg() -> i32 {
        if is_connect_state_true() {
            1
        } else {
            0
        }
    }
    /// disconnect from websocket server
    fn disconnect_sg() -> SDKApiResult {
        match SG_WS_SDK_INSTANCE.try_get() {
            Some(instance) => {
                return instance.read().unwrap().ws_instance.disconnect();
            }
            None => {
                return SDKApiResult::from(anyhow!(SDKResultError::ErrorLinkSdkNotInited));
            }
        };
    }
}
